package com.stqf.scala.flink;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

/**
 * Created By admin
 * Date: 2020/5/11
 * Description:
 */
public class RecvKafkaStreaming {

    private static final Logger log = LoggerFactory.getLogger(RecvKafkaStreaming.class);

    public static void main(String[] args) throws Exception {

        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "192.168.2.89:9092");
        properties.setProperty("auto.offset.reset", "earliest");

        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("os-storage-snapFace", new SimpleStringSchema(), properties);
        DataStream<String> dataStream = env.addSource(consumer);


        dataStream.addSink(new RichSinkFunction<String>() {
            @Override
            public void invoke(String value, Context context) throws Exception {
                log.info("Invoke ... {}", value);
            }
        });

        env.execute("Flink Kafka");


    }

}
